/**
 * Copyright 2014 opencxa.org
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.cxa.rpc.session;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.cxa.exec.updater.Updatable;
import org.cxa.misc.ExceptionUtils;
import org.cxa.rpc.messages.KeepAliveResponseRpcMessage;
import org.cxa.rpc.messages.KeepAliveRpcMessage;
import org.cxa.rpc.messages.NotificationRpcMessage;
import org.cxa.rpc.messages.RequestRpcMessage;
import org.cxa.rpc.messages.ResponseRpcMessage;
import org.cxa.rpc.messages.RpcMessage;
import org.cxa.rpc.provider.RpcException;
import org.cxa.rpc.provider.RpcException.RpcError;
import org.cxa.rpc.transportChannel.TransportChannel;
import org.cxa.rpc.transportChannel.TransportChannel.DisconnectReason;
import org.cxa.rpc.transportChannel.TransportChannel.TransportChannelListener;
import org.cxa.stateMachine.State;
import org.cxa.stateMachine.StateMachine;
import org.cxa.timeUtils.TimeDiff;


/**
 * @author Christopher Armenio
 */
public class SessionManager implements Updatable
{	
	private static final int MAX_ID = 65535;
	
	private static final int SESSION_STALE_PERIOD_MS = 10000;
	private static final int KEEP_ALIVE_SEND_INTERVAL_MS = SESSION_STALE_PERIOD_MS;
	public static final int SESSION_TIMEOUT_PERIOD_MS = SESSION_STALE_PERIOD_MS * 5;
	
	private static final int DEFAULT_SYNC_TIMEOUT_MS = 1000;
	private static final int DEFAULT_ASYNC_TIMEOUT_MS = 1000;
	
	private static final int INFLIGHT_REQUEST_CLEANUP_PERIOD_MS = DEFAULT_SYNC_TIMEOUT_MS * 5;
	
	
	public interface SessionListener
	{
		public abstract void onSessionOpen();
		public abstract void onSessionClose(DisconnectReason reasonIn);
		
		public abstract void onRequestReceived(RequestRpcMessage msgIn);
		public abstract void onNotificationReceived(NotificationRpcMessage msgIn);
		public abstract void onUnmatchedResponseReceived(ResponseRpcMessage msgIn);
	}
	
	
	public static abstract class SessionListenerAdapter implements SessionListener
	{
		public void onSessionOpen() {}
		public void onSessionClose(DisconnectReason reasonIn) {}
		
		public void onRequestReceived(RequestRpcMessage msgIn) {}
		public void onNotificationReceived(NotificationRpcMessage msgIn) {}
		public void onUnmatchedResponseReceived(ResponseRpcMessage msgIn) {}
	}
	
	
	public interface AsynchronousRequestListener
	{
		public abstract void onResponseReceived(ResponseRpcMessage msgIn);
	}
	
	
	private enum SessionState
	{
		STATE_CLOSED,
		STATE_OPEN,
		STATE_OPEN_STALE
	}
	
	
	private class InflightRequest
	{
		ResponseRpcMessage response = null;
		
		Semaphore sem = null;
		
		AsynchronousRequestListener asyncListener = null;
		TimeDiff td_asyncExpiration = new TimeDiff();
		
		private InflightRequest(Semaphore semIn) { this.sem = semIn; }
		private InflightRequest(AsynchronousRequestListener asyncListenerIn) { this.asyncListener = asyncListenerIn; } 
	}
	
	
	private final TransportChannel transportChannel;
	private final List<SessionListener> listeners = new ArrayList<SessionListener>();
	
	private final TimeDiff td_lastMsgRx = new TimeDiff();
	
	private final Logger logger;
	private final StateMachine<SessionState> stateMachine = new StateMachine<SessionState>(this.toString());
	private DisconnectReason lastDisconnectReason = null;
	
	private int nextId = 0;
	private Map<Integer, InflightRequest> inflightRequests = Collections.synchronizedMap(new HashMap<Integer, InflightRequest>());
	private final TimeDiff td_inflightReqCleanup = new TimeDiff();
	
	private boolean debug_stopRespondingToKeepAlives = false;
	
	
	public SessionManager(TransportChannel tcIn) throws IllegalArgumentException
	{		
		// save our references
		this.transportChannel = tcIn;
		
		// setup our logger
		this.logger = LogManager.getLogger(this.toString());
	
		// setup our TransportChannelListener
		this.logger.debug("owns " + this.transportChannel.toString() );
		this.transportChannel.addTransportChannelListener(new TransportChannelListener()
		{
			@Override
			public void onConnect()
			{
				if( stateMachine.getCurrentState() == SessionState.STATE_CLOSED ) stateMachine.transition(SessionState.STATE_OPEN);
			}

			
			@Override
			public void onDisconnect(DisconnectReason reasonIn)
			{
				lastDisconnectReason = reasonIn;
				stateMachine.transition(SessionState.STATE_CLOSED);
			}

			
			@Override
			public void onMessageReceived(RpcMessage msgIn)
			{
				if( stateMachine.getCurrentState() == SessionState.STATE_CLOSED ) return;
				
				td_lastMsgRx.setStartTime_now();
				
				switch(msgIn.getMsgType())
				{
					case KEEPALIVE:
						// send our response
						if( debug_stopRespondingToKeepAlives) transportChannel.sendMessage(new KeepAliveResponseRpcMessage());
						break;
						
					case KEEPALIVE_RESPONSE:
						// nothing to do here
						break;
						
					case NOTIFICATION:
						// notify our listeners
						for( SessionListener currListener : listeners )
						{
							currListener.onNotificationReceived((NotificationRpcMessage)msgIn);
						}
						break;
						
					case REQUEST:
						// notify our listeners
						for( SessionListener currListener : listeners )
						{
							currListener.onRequestReceived((RequestRpcMessage)msgIn);
						}
						break;
						
					case RESPONSE:
						ResponseRpcMessage msgIn_resp = (ResponseRpcMessage)msgIn;
						InflightRequest ifr = inflightRequests.get(msgIn_resp.getId());
						if( ifr == null )
						{
							// notify our listeners
							for( SessionListener currListener : listeners )
							{
								currListener.onUnmatchedResponseReceived(msgIn_resp);
							}
							return;
						}
						
						// if we made it here, it _is_ a response we're waiting for
						ifr.response = msgIn_resp;
						
						// synchronous response
						if( ifr.sem != null ) ifr.sem.release();
						
						// asynchronous response
						if( ifr.asyncListener != null )
						try { ifr.asyncListener.onResponseReceived(msgIn_resp); }
						catch( Exception e ) { logger.warn(ExceptionUtils.getExceptionLogString(e, ExceptionUtils.MSG_OPERATION_WILL_CONTINUE)); }
							
						break;
						
					default:
						logger.warn("unknown message type received");
						break;
					
				}
			}
		});
		
		
		// setup our state machine
		this.stateMachine.addState(new State<SessionState>(SessionState.STATE_CLOSED)
		{
			@Override
			public void enter(SessionState prevStateIn)
			{
				// if there was no previous state, we must just be starting...
				if( prevStateIn == null ) return;
				
				// see if our transport is still connected
				if( getTransportChannel().isConnected() ) logger.warn("session is closed but transport is still open");
				
				// if we made it here, we must have been in another state...notify our listeners
				for( SessionListener currListener : listeners )
				{
					try { currListener.onSessionClose((lastDisconnectReason == null) ? DisconnectReason.UNKNOWN : lastDisconnectReason); }
					catch(Exception e) { logger.warn(ExceptionUtils.getExceptionLogString(e, ExceptionUtils.MSG_OPERATION_WILL_CONTINUE)); }
				}
				lastDisconnectReason = null;
			}
			
			@Override
			public void state()
			{
				// make sure we haven't magically become connected...
				if( getTransportChannel().isConnected() ) stateMachine.transition(SessionState.STATE_OPEN);
			}
		});
		
		this.stateMachine.addState(new State<SessionState>(SessionState.STATE_OPEN)
		{	
			@Override
			public void enter(SessionState prevStateIn)
			{
				// if we were previously closed, notify our listeners
				if( prevStateIn == SessionState.STATE_CLOSED )
				{
					for( SessionListener currListener : listeners )
					{
						try{ currListener.onSessionOpen(); }
						catch(Exception e) { logger.warn(ExceptionUtils.getExceptionLogString(e, ExceptionUtils.MSG_OPERATION_WILL_CONTINUE)); }
					}
					
					// reset our lastMsgRx timeout and clear our inflight requests
					td_lastMsgRx.setStartTime_now();
					inflightRequests.clear();
				}
			}
			
			@Override
			public void state()
			{
				// see if we've become stale
				if( td_lastMsgRx.isElapsed(SESSION_STALE_PERIOD_MS, TimeUnit.MILLISECONDS) )
				{
					stateMachine.transition(SessionState.STATE_OPEN_STALE);
					return;
				}
				
				if( td_inflightReqCleanup.isElapsed_recurring(INFLIGHT_REQUEST_CLEANUP_PERIOD_MS, TimeUnit.MILLISECONDS) ) cleanupInflightRequests();
			}
		});
		
		this.stateMachine.addState(new State<SessionState>(SessionState.STATE_OPEN_STALE)
		{
			private final TimeDiff td_sendKeepAlive = new TimeDiff();
			private final TimeDiff td_sessionTimeout = new TimeDiff();
			
			@Override
			public void enter(SessionState prevStateIn)
			{
				// send a keepalive NOW
				// (if it fails, it should be handled by transport channel)
				transportChannel.sendMessage(new KeepAliveRpcMessage());
				
				// reset our timeouts
				this.td_sendKeepAlive.setStartTime_now();
				this.td_sessionTimeout.setStartTime_now();
			}
			
			@Override
			public void state()
			{
				if( !td_lastMsgRx.isElapsed(SESSION_STALE_PERIOD_MS, TimeUnit.MILLISECONDS) )
				{
					// we've received a message...we're no longer stale
					stateMachine.transition(SessionState.STATE_OPEN);
					return;
				}
				if( this.td_sessionTimeout.isElapsed(SESSION_TIMEOUT_PERIOD_MS, TimeUnit.MILLISECONDS) )
				{
					logger.warn("session has become non-responsive, disconnecting");
					getTransportChannel().disconnect(DisconnectReason.KEEPALIVE_FAILED);
					// transport channel listener will handle the rest from here...
				}
				else if( this.td_sendKeepAlive.isElapsed_recurring(KEEP_ALIVE_SEND_INTERVAL_MS, TimeUnit.MILLISECONDS) )
				{
					// send a keepalive NOW
					// (if it fails, it should be handled by transport channel)
					transportChannel.sendMessage(new KeepAliveRpcMessage());
				}
				
				if( td_inflightReqCleanup.isElapsed_recurring(INFLIGHT_REQUEST_CLEANUP_PERIOD_MS, TimeUnit.MILLISECONDS) ) cleanupInflightRequests();
			}
		});
		this.stateMachine.transition_now(SessionState.STATE_CLOSED);
	}
	
	
	/**
	 * Adds a {@link SessionManagerListener} to the SessionManager that will
	 * be notified of various events within the SessionManager.
	 * 
	 * @param slIn the SessionListener
	 */
	public void addSessionListener(SessionListener slIn)
	{
		if( slIn == null ) throw new NullPointerException("listener must NOT be null");
		this.listeners.add(slIn);
	}
	
	
	public void removeAllSessionListeners()
	{
		// remove everything from our list
		this.listeners.clear();
	}
	
	
	public synchronized ResponseRpcMessage sendRequest(RequestRpcMessage msgIn)
	{
		// set the id of our message (assuming we don't already have one)
		if( msgIn.getId() == null ) msgIn.setId(++this.nextId);
		if( this.nextId == SessionManager.MAX_ID ) this.nextId = 0;
		
		// setup our inflight request so we can be ready for a response
		Semaphore syncRxSemaphore = new Semaphore(0);
		this.inflightRequests.put(msgIn.getId(), new InflightRequest(syncRxSemaphore));
		
		// send our message
		// note: if an exception occurred in sendMessage, it will report an
		// unexpectedTransportClosure and will be handled accordingly
		if( !this.transportChannel.sendMessage(msgIn) )
		{
			// message send failed
			this.logger.error("failed to send request");
			
			this.inflightRequests.remove(msgIn.getId());
			return null;
		}
		
		// message was sent successfully...now wait for our response (or a timeout)
		try
		{
			syncRxSemaphore.tryAcquire(DEFAULT_SYNC_TIMEOUT_MS, TimeUnit.MILLISECONDS);
		}
		catch (InterruptedException e) { this.logger.warn(ExceptionUtils.getExceptionLogString(e, ExceptionUtils.MSG_OPERATION_WILL_CONTINUE)); }
		
		
		// one way or another, we're done waiting for the response...
		// get our request and return the buffer
		InflightRequest origRequest = this.inflightRequests.remove(msgIn.getId());
		return (origRequest.response != null) ? origRequest.response : new ResponseRpcMessage(msgIn, RpcException.exceptionFromError(RpcError.REQUEST_TIMEOUT));
	}
	
	
	public synchronized boolean sendRequest(RequestRpcMessage msgIn, AsynchronousRequestListener listenerIn)
	{
		// set the id of our message (assuming we don't already have one)
		if( msgIn.getId() == null ) msgIn.setId(++this.nextId);
		if( this.nextId == SessionManager.MAX_ID ) this.nextId = 0;
		
		this.inflightRequests.put( msgIn.getId(), new InflightRequest(listenerIn) );
		
		// send our message
		// note: if an exception occurred in sendMessage, it will report an
		// unexpectedTransportClosure and will be handled accordingly
		if( !this.transportChannel.sendMessage(msgIn) )
		{
			// message send failed
			this.logger.error("failed to send request");
			
			this.inflightRequests.remove(msgIn.getId());
			return false;
		}
		
		return true;
	}
	
	
	public boolean sendMessage(RpcMessage msgIn)
	{
		return this.transportChannel.sendMessage(msgIn);
	}
	
	
	public TransportChannel getTransportChannel()
	{
		return this.transportChannel;
	}
	
	
	public void debug_stopRespondingToKeepAlives()
	{
		this.debug_stopRespondingToKeepAlives = true;
	}
	
	
	private void cleanupInflightRequests()
	{
		Iterator<Entry<Integer, InflightRequest>> it = this.inflightRequests.entrySet().iterator();
		while( it.hasNext() )
		{
			InflightRequest currIfr = it.next().getValue();
			// only asynchronous requests
			if( (currIfr.sem == null) && currIfr.td_asyncExpiration.isElapsed(DEFAULT_ASYNC_TIMEOUT_MS, TimeUnit.MILLISECONDS) ) it.remove();
		}
	}
	
	
	@Override
	public boolean start()
	{
		return (this.stateMachine.getCurrentState() == SessionState.STATE_CLOSED);
	}


	@Override
	public void update()
	{
		this.stateMachine.update();
	}


	@Override
	public void stop()
	{
		if( this.stateMachine.getCurrentState() != SessionState.STATE_CLOSED )
		{
			// transition now since update won't get called again
			this.stateMachine.transition_now(SessionState.STATE_CLOSED);
		}
	}
	
	
	/*
	 * (non-Javadoc)
	 * @see java.lang.Object#toString()
	 */
	@Override
	public String toString()
	{
		return String.format("%s.%s", this.getClass().getSimpleName(), System.identityHashCode(this));
	}
}
